package com.lunsheng.train

import java.sql.{Connection, DriverManager}

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * @Description
  * @author zhonglunsheng
  * @create 2020-02-10 23:20
  */
object WindowWorldCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[3]").setAppName("WorldCountV2")
    val ssc = new StreamingContext(conf, Seconds(5))
    ssc.checkpoint("E:/temp/checkpoint")
    val lines = ssc.textFileStream("D:/temp")

    val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

    val windowResult = result.reduceByKeyAndWindow((a:Int,b:Int) => a + b, Seconds(30), Seconds(10))

    windowResult.print()
    windowResult.foreachRDD { rdd =>
      rdd.foreachPartition { partitionOfRecords =>
        val connection = createNewConnection()
        partitionOfRecords.foreach(record =>
          if (!isExist(record._1, connection)) {
            connection.createStatement()
              .execute("insert into word(word, count) values('"+ record._1+"', '"+record._2+"')")
          }else {
            connection.createStatement()
              .execute("update word set count = '"+ record._2 +"' where word = '"+ record._1 +"'")
          }
        )
        connection.close()
      }
    }
    ssc.start()
    ssc.awaitTermination()
  }

  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val currentValue = newValues.sum
    val preValue = runningCount.getOrElse(0)
    val newCount = currentValue + preValue
    Some(newCount)
  }

  def createNewConnection(): Connection = {
    Class.forName("com.mysql.jdbc.Driver")
    DriverManager.getConnection("jdbc:mysql://localhost:3306/s_streaming", "root", "123456")
  }

  def isExist(word: String, connection: Connection):Boolean = {
    var result = false
    val resultSet = connection.createStatement().executeQuery("select count(*) as count from word where word = '"+ word +"'")
    while(resultSet.next()) {
      result = resultSet.getInt("count") > 0
    }
    result
  }
}
